package kernel_db.db;

import java.sql.*;
import java.util.*;

import common.entities.enums.ACTION;
import common.entities.enums.ENTITIES;
import common.entities.enums.PATTERN;
import common.entities.enums.SEX;
import common.entities.nonpersistent.GameQuestion;
import common.entities.persistent.*;
import kernel_db.interfaces.*;
import kernel_db.parser.ParserDB;

/**
 * @author Dan boy
 * 
 */
public class DB implements IDB {

	private Queue<Connection> connection_pool; // DB connection pool
	private String path;
	private String user;
	private String password;
	private static final int numOfLevels = 3;
	private static final int numOfQuestionsPerLevel = 6;
	private int movieMinYear;
	private int movieMaxYear;
	private ParserDB parser;
	private int EasyLevelYear = 2000;
	private int MediumLevelYear = 1990;
	private boolean isDriverLoaded;

	/**
	 * Empty constructor
	 */
	public DB(String path, String user, String password, int movieMaxYear, int maxBatch, int fetchSize) {
		this.path = path;
		this.user = user;
		this.password = password;
		this.movieMinYear = Movie.releaseYearMinValue;
		this.movieMaxYear = movieMaxYear;
		isDriverLoaded = false;
		connection_pool = new LinkedList<Connection>();
		parser = new ParserDB(this, maxBatch, fetchSize);
	}

	/**
	 * This function returns a connection from the connection pool or creates a
	 * new connection if connection pool is empty
	 * 
	 * @return the created connection
	 * @throws DB_Error
	 */
	public synchronized Connection dequeueConnection() throws DB_Error {

		Connection conn = connection_pool.poll();

		if (conn != null)
			return conn;

		// if we arrive here then pool is empty
		// and we're going to construct a new connection

		if (!isDriverLoaded) {
			// loading the driver
			try {
				Class.forName("com.mysql.jdbc.Driver");
			} catch (ClassNotFoundException e) {
				throw new DB_Error("Unable to load the MySQL JDBC driver.");
			}
			// driver loaded successfully
		}

		// creating the connection
		try {
			conn = DriverManager.getConnection(path, user, password);
			return conn;
		} catch (SQLException e) {
			throw new DB_Error("Unable to connect - " + e.getMessage());
		}
		// connected to db successfully
	}

	/**
	 * puts the connection back into the connection pool
	 * 
	 * @throws DB_Error
	 */
	public void enqueueConnection(Connection conn) {
		connection_pool.add(conn);
	}

	/**
	 * close all connections in connection queue use this function when process
	 * has finished
	 * 
	 * @throws DB_Error
	 */
	public synchronized void closeAllConnections() throws DB_Error {

		try {
			for (Connection conn : connection_pool) {
				conn.close();// closing the connection
			}
		} catch (SQLException e) {
			throw new DB_Error("Unable to close the connection - "
					+ e.getMessage());
		}

	}

	/**
	 * create new user
	 * 
	 * @throws DB_Error
	 *             if user already exist
	 */
	public synchronized void createUser(User user) throws DB_Error {
		Connection conn = dequeueConnection();
		Statement stmt;
		ResultSet rs;

		try {
			stmt = conn.createStatement();
			// check if name already exist
			rs = stmt.executeQuery("SELECT name FROM user WHERE lower(name)='"
					+ user.escapedGetName().toLowerCase() + "'");
			if (rs.next()) {
				enqueueConnection(conn);
				rs.close();
				stmt.close();
				throw new DB_Error("Name already exist", false);
			}
			rs.close();

			// insert new user to database
			stmt.executeUpdate("INSERT INTO user(name, password) VALUES('"
					+ user.escapedGetName() + "','" + user.escapedGetPassword() + "')");

			stmt.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR creating new user - " + e.getMessage());
		}
		enqueueConnection(conn);
	}

	/**
	 * authenticate user
	 * 
	 * @return new instance of User with the user's details (or null if user
	 *         doesn't exist/incorrect password)
	 * @throws DB_Error
	 */
	public synchronized User authenticateUser(User userOld) throws DB_Error {
		String userName = userOld.escapedGetName(), password = userOld.getPassword();
		Connection conn = dequeueConnection();
		Statement stmt;
		ResultSet rs;
		User user = null;

		try {
			stmt = conn.createStatement();

			rs = stmt.executeQuery("SELECT * FROM user WHERE lower(name)='"
					+ userName.toLowerCase() + "'");
			if (rs.next()) {
				if (password.equals(rs.getString(3)))
					user = new User(rs.getInt(1), rs.getString(2),
							rs.getString(3), rs.getInt(4), rs.getLong(5),
							rs.getInt(6));
			}

			rs.close();
			stmt.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR authenticate user - " + e.getMessage());
		}
		enqueueConnection(conn);
		return user;
	}

	/**
	 * 
	 * @return a list of top players
	 */
	@Override
	public synchronized List<User> hallOfFame(int numberOfUsers)
			throws DB_Error {
		Connection conn = dequeueConnection();
		Statement stmt;
		ResultSet rs;
		List<User> users = new LinkedList<User>();
		try {
			stmt = conn.createStatement();
			rs = stmt
					.executeQuery("SELECT * FROM user ORDER BY highest_win DESC, money_won DESC LIMIT "
							+ numberOfUsers);
			while (rs.next()) {
				User user = new User(rs.getInt(1), rs.getString(2),
						rs.getString(3), rs.getInt(4), rs.getLong(5),
						rs.getInt(6));
				users.add(user);
			}
			rs.close();
			stmt.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR hallOfFame - " + e.getMessage());
		}
		enqueueConnection(conn);
		return users;
	}

	/**
	 * counts number of questions per level
	 * 
	 * @param table
	 * @param numOfQuestions
	 *            - total number of questions per level
	 * @param patternNumOfQuestions
	 *            - number of questions per level in a specific pattern
	 * @param stmt
	 * @param conn
	 * @throws DB_Error
	 */
	private void countQuestions(String table, int[] numOfQuestions, int[] patternNumOfQuestions) throws DB_Error {
		Connection conn = dequeueConnection();
		Statement stmt;
		ResultSet rs;
		try {
			stmt = conn.createStatement();
			for (int i = 1; i <= numOfLevels; i++) {
				rs = stmt.executeQuery("SELECT count(*) FROM " + table
						+ " WHERE level=" + i);
				rs.next();
				int countReturned = rs.getInt(1);
				numOfQuestions[i - 1] += countReturned;
				patternNumOfQuestions[i - 1] += countReturned;
				rs.close();
			}
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR creating game - " + e.getMessage());
		}
		enqueueConnection(conn);
	}

	/**
	 * select k random numbers from [1..n]
	 * 
	 * @param k
	 *            - number of randoms to choose
	 * @param n
	 *            - upper bound of the numbers to choose from
	 * @return a list with the k chosen numbers
	 */
	private List<Integer> generateKRandumNumbers(int k, int n) {
		if (n < 1 || k < 1 || k > n)
			return null;
		List<Integer> randoms = new ArrayList<Integer>();
		Map<Integer, Integer> chosen = new HashMap<Integer, Integer>();
		Random r = new Random();

		for (int i = 0; i < k; i++) {
			int index = r.nextInt(n);
			if (chosen.containsKey(index)) {
				randoms.add(chosen.get(index) + 1);
				if (chosen.containsKey(n - 1))
					chosen.put(index, chosen.get(n - 1));
				else
					chosen.put(index, n - 1);
			} else {
				randoms.add(index + 1);
				if (chosen.containsKey(n - 1))
					chosen.put(index, chosen.get(n - 1));
				else
					chosen.put(index, n - 1);
			}
			n--;
		}

		return randoms;
	}

	private PATTERN intToPattern(int x) {
		switch (x) {
		case 0:
			return PATTERN.ACTOR_PLAYED_IN_MOVIE;
		case 1:
			return PATTERN.GENRE_OF_MOVIE;
		case 2:
			return PATTERN.MOVIE_ACTOR_IN;
		case 3:
			return PATTERN.MOVIE_DIRECTED_BY;
		case 4:
			return PATTERN.MOVIE_RELEASED_IN_YEAR;
		default:
			return null;
		}
	}

	/**
	 * choose how many questions to select from each pattern and level
	 * 
	 * @param patternNumOfQuestions
	 *            - number of questions currently in database divided by
	 *            patterns and levels
	 * @return
	 */
	private int[][] questionsToChoose(int[][] patternNumOfQuestions,
			int[] totalNumOfQuestions) {
		int[][] patternNumOfQuestionsToChoose = new int[PATTERN
				.getNumOfPatterns()][numOfLevels];

		for (int i = 0; i < numOfLevels; i++) {
			List<Integer> radomNumbers = generateKRandumNumbers(
					numOfQuestionsPerLevel, totalNumOfQuestions[i]);

			for (int x : radomNumbers) {
				for (int j = 0; j < PATTERN.getNumOfPatterns(); j++) {
					if (x <= patternNumOfQuestions[j][i]) {
						patternNumOfQuestionsToChoose[j][i]++;
						break;
					} else {
						x -= patternNumOfQuestions[j][i];
					}
				}
			}
		}

		return patternNumOfQuestionsToChoose;
	}

	private void addQuestion(List<GameQuestion> questions, PATTERN pattern,
			int level, int numberOfQuestions) throws DB_Error {
		Connection conn = dequeueConnection();
		Statement stmt1, stmt2;
		ResultSet rs1, rs2;

		String answerTable = "";
		String questionTable = "";
		switch (pattern) {
		case ACTOR_PLAYED_IN_MOVIE:
			answerTable = "actor";
			questionTable = ",movie WHERE movie_id=movie.id AND ";
			break;
		case GENRE_OF_MOVIE:
			answerTable = "genre";
			questionTable = ",movie WHERE movie_id=movie.id AND ";
			break;
		case MOVIE_RELEASED_IN_YEAR:
			answerTable = "movie";
			questionTable = " WHERE ";
			break;
		case MOVIE_ACTOR_IN:
			answerTable = "movie";
			questionTable = ",actor WHERE actor_id=actor.id AND ";
			break;
		case MOVIE_DIRECTED_BY:
			answerTable = "movie";
			questionTable = ",director WHERE director_id=director.id AND ";
			break;
		}
		String r = String.format("ans_%s_r", answerTable);
		String w1 = String.format("ans_%s_w1", answerTable);
		String w2 = String.format("ans_%s_w2", answerTable);
		String w3 = String.format("ans_%s_w3", answerTable);

		try {
			stmt1 = conn.createStatement();
			stmt2 = conn.createStatement();

			rs1 = stmt1.executeQuery("SELECT * FROM " + pattern.getTableName()
					+ questionTable + "level=" + level
					+ " ORDER BY RAND() LIMIT " + numberOfQuestions);

			while (rs1.next()) {
				rs2 = stmt2.executeQuery("SELECT * FROM " + answerTable
						+ " WHERE id IN (" + rs1.getInt(r) + ", "
						+ rs1.getInt(w1) + ", " + rs1.getInt(w2) + ", "
						+ rs1.getInt(w3) + ")");

				String questionName = "";
				if (pattern.equals(PATTERN.MOVIE_RELEASED_IN_YEAR))
					questionName = Integer.toString(rs1.getInt("year"));
				else if (pattern.equals(PATTERN.MOVIE_DIRECTED_BY)
						|| pattern.equals(PATTERN.MOVIE_ACTOR_IN))
					questionName = rs1.getString("first_name") + " "
							+ rs1.getString("last_name");
				else
					questionName = rs1.getString("name");

				// setting the answers
				String[] answers = new String[4];
				int rightAnswerIndex = -1;
				int i = 0;
				while (rs2.next()) {
					if (rs2.getInt("id") == rs1.getInt(r)) {
						rightAnswerIndex = i;
					}
					if (answerTable.equals("actor")) {
						answers[i] = rs2.getString("first_name");
						answers[i] += " " + rs2.getString("last_name");
					} else {
						answers[i] = rs2.getString("name");
					}
					i++;
				}
				rs2.close();

				questions.add(new GameQuestion(pattern
						.fillQuestion(questionName), rightAnswerIndex,
				/* questionValue */0,// kernel sets this value
						answers));
			}
			rs1.close();
			stmt2.close();
			stmt1.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR add question to game - " + e.getMessage());
		}
		enqueueConnection(conn);
	}

	// return a list of game questions
	// 5 easy, 5 inter, 5 hard, 3 extra for each level at this order
	public synchronized List<GameQuestion> generateGame() throws DB_Error {
		int[] totalNumOfQuestions = new int[numOfLevels];
		int[][] patternNumOfQuestions = new int[PATTERN.getNumOfPatterns()][numOfLevels];
		List<GameQuestion> game = new ArrayList<GameQuestion>();

		for (int i = 0; i < PATTERN.getNumOfPatterns(); i++) {
			countQuestions(intToPattern(i).getTableName(), totalNumOfQuestions,
					patternNumOfQuestions[i]);
		}

		for (int i = 0; i < numOfLevels; i++) {
			if (totalNumOfQuestions[i] < numOfQuestionsPerLevel) {
				throw new DB_Error(
						"There are not enough questions to start a new game.",false);
			}
		}

		// determine how many questions to choose from each patter at each level
		patternNumOfQuestions = questionsToChoose(patternNumOfQuestions,
				totalNumOfQuestions);

		for (int i = 0; i < numOfLevels; i++) {
			List<GameQuestion> questions = new ArrayList<GameQuestion>();
			for (int j = 0; j < PATTERN.getNumOfPatterns(); j++) {
				if (patternNumOfQuestions[j][i] == 0)
					continue;
				addQuestion(questions, intToPattern(j), i + 1,
						patternNumOfQuestions[j][i]);
			}
			Collections.shuffle(questions);
			game.addAll(questions);
		}
		return game;
	}

	/**
	 * update user's score
	 * 
	 * @throws DB_Error
	 */
	@Override
	public synchronized void updateUserScore(User user) throws DB_Error {
		Connection conn = dequeueConnection();
		Statement stmt;

		try {
			stmt = conn.createStatement();

			// updating user's score
			stmt.executeUpdate("UPDATE user SET games_played="
					+ user.getGamesPlayed() + ", money_won="
					+ user.getMoneyWon() + ", highest_win="
					+ user.getPersonalRecord() + " WHERE id=" + user.getId());

			stmt.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR updating user's info - " + e.getMessage());
		}
		enqueueConnection(conn);
	}

	@Override
	public synchronized void actionUser(ACTION action, User user) throws DB_Error {
		actionForEntity(
				action,
				"user",
				new String[] { "name", "password", "games_played", "money_won",
						"highest_win" },
				new String[] { "'" + user.escapedGetName() + "'",
						"'" + user.escapedGetPassword() + "'",
						Integer.toString(user.getGamesPlayed()),
						Long.toString(user.getMoneyWon()),
						Integer.toString(user.getPersonalRecord()) }, 1, "id",
				user.getId());
	}

	@Override
	public synchronized int countUsers(String userNameSubStr) throws DB_Error {
		String[] Table = new String[] { "user" };
		if (userNameSubStr == null || userNameSubStr.length() == 0){
			return countGeneric(Table, new String[0], new String[0], "id");
		}	
		return countGeneric(Table, new String[] { "lower(name) LIKE " },
				new String[] { "'%" + userNameSubStr + "%'" }, "id");
	}

	@Override
	public synchronized List<User> getUsers(String userNameSubStr, int from, int to)
			throws DB_Error {
		return Selector.getUsers(userNameSubStr, from, to, this);
	}

	@Override
	public synchronized void actionGenre(ACTION action, Genre genre) throws DB_Error {
		actionForEntity(action, "genre", new String[] { "name" },
				new String[] { "'" + genre.escapedGetName() + "'" }, 1, "id",
				genre.getId());
	}

	@Override
	public synchronized int countGenres(String GenreSubStr, Movie movie) throws DB_Error {
		ArrayList<String> tables = new ArrayList<String>();
		ArrayList<String> params = new ArrayList<String>();
		ArrayList<String> args = new ArrayList<String>();
		
		tables.add("genre");
		
		if (GenreSubStr != null && GenreSubStr.length() > 0) {
			params.add("lower(name) LIKE ");
			args.add("'%" + GenreSubStr + "%'");
		}
		
		if (movie != null) {
			tables.add("genre_movie");
			params.add("genre_id = "); args.add("id");
			params.add("movie_id = "); args.add(Integer.toString(movie.getId()));
		}

		return countGeneric(
				ListToArray(tables), 
				ListToArray(params),
				ListToArray(args), 
				"id");
	}

	@Override
	public synchronized List<Genre> getGenres(String GenreSubStr, int from, int to)
			throws DB_Error {
		return Selector.getGenres(GenreSubStr, from, to, this);
	}

	private synchronized void actionForEntity(ACTION action, String table,
			String[] params, String[] values, int keyNum, String idName, int id)
			throws DB_Error {
		switch (action) {
		case CREATE:
			insertGeneric(table, params, values, keyNum);
			break;
		case DELETE:
			deleteGeneric(table, new String[] { idName }, new int[] { id });
			break;
		case UPDATE:
			updateGeneric(table, params, values, keyNum, idName, id);
			break;
		}
	}

	private synchronized void deleteGeneric(String table, String[] keyNames,
			int[] keyValues) throws DB_Error {
		Connection conn = dequeueConnection();
		Statement stmt;

		try {
			stmt = conn.createStatement();

			String k = "";
			for (int i = 0; i < keyNames.length; i++) {
				k += keyNames[i] + "=" + keyValues[i];

				if (i != keyNames.length - 1)
					k += " AND ";
			}

			stmt.executeUpdate("DELETE FROM " + table + " WHERE " + k);

			stmt.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR deleting from DB - " + e.getMessage());
		}
		enqueueConnection(conn);
	}

	/**
	 * insert new tuple to DB
	 * 
	 * @param table
	 *            - name of the table to insert to
	 * @param param
	 *            - name of the columns in the table to insert values
	 * @param values
	 *            - the values to insert
	 * @param keySize
	 *            - first keySize parameters of the param list need to be the
	 *            primary key of the table
	 * @throws DB_Error
	 */
	private synchronized void insertGeneric(String table, String[] param,
			String[] values, int keySize) throws DB_Error {
		Connection conn = dequeueConnection();
		Statement stmt;
		ResultSet rs;

		try {
			stmt = conn.createStatement();

			// check if tuple already exist
			String selectQuery = "SELECT * FROM " + table + " WHERE ";
			for (int i = 0; i < keySize; i++) {
				selectQuery += "lower(" + param[i] + ")="
						+ values[i].toLowerCase();
				if (i != keySize - 1) {
					selectQuery += " AND ";
				}
			}
			rs = stmt.executeQuery(selectQuery);
			if (rs.next()) {
				enqueueConnection(conn);
				rs.close();
				stmt.close();
				throw new DB_Error("Entry already exist", false);
			}
			rs.close();

			// insert new tuple to the database
			String p = "(";
			String v = "(";
			for (int i = 0; i < param.length; i++) {
				p += param[i];
				v += values[i];

				if (i == param.length - 1) {
					if (table.compareTo("movie") == 0){
						p += ",top_movie";
						v += ",1";
					}
					else if (table.compareTo("actor") == 0){
						p += ",top_actor";
						v += ",1";
					}
					p += ")";
					v += ")";
				} else {
					p += ",";
					v += ",";
				}
			}

			stmt.executeUpdate("INSERT INTO " + table + " " + p + " VALUES "
					+ v);

			stmt.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR inserting item to DB - " + e.getMessage());
		}
		enqueueConnection(conn);
	}

	/**
	 * update a tuple in the DB
	 * 
	 * @throws DB_Error
	 */
	private synchronized void updateGeneric(String table, String[] param,
			String[] values, int keySize, String keyName, int keyValue)
			throws DB_Error {
		Connection conn = dequeueConnection();
		Statement stmt;
		ResultSet rs;

		String p = "";
		for (int i = 0; i < param.length; i++) {
			p += param[i] + "=" + values[i];

			if (i != param.length - 1)
				p += ",";
		}

		String k = keyName + "=" + keyValue;
		String selectQuery = "SELECT * FROM " + table + " WHERE ";
		for (int i = 0; i < keySize; i++) {
			selectQuery += "lower(" + param[i] + ")=" + values[i].toLowerCase();
			if (i != keySize - 1) {
				selectQuery += " AND ";
			}
		}
		selectQuery += " AND NOT(" + k + ")";

		try {
			// check if tuple already exist
			stmt = conn.createStatement();
			rs = stmt.executeQuery(selectQuery);
			if (rs.next()) {
				enqueueConnection(conn);
				rs.close();
				stmt.close();
				throw new DB_Error(
						"Can't update entry with the same unique information that already exist",
						false);
			}
			rs.close();

			// update
			stmt.executeUpdate("UPDATE " + table + " SET " + p + " WHERE " + k);
			stmt.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR inserting item to DB - " + e.getMessage());
		}
		enqueueConnection(conn);
	}

	private synchronized int countGeneric(String[] tables, String[] param,
			String[] values, String fieldName) throws DB_Error {
		Connection conn = dequeueConnection();
		Statement stmt;
		ResultSet rs;
		int count = 0;

		// create query
		String fromTables = "";
		for (int i = 0; i < tables.length; i++) {
			fromTables += tables[i];
			if (i + 1 < tables.length) {
				fromTables += ",";
			}
		}
		
		String selectQuery = "SELECT count( " + tables[0] + "." + fieldName
				+ ") FROM " + fromTables;
		for (int i = 0; i < param.length; i++) {
			if (i == 0)
				selectQuery += " WHERE ";
			selectQuery += param[i] + values[i].toLowerCase();
			if (i != param.length - 1) {
				selectQuery += " AND ";
			}
		}

		try {
			stmt = conn.createStatement();
			rs = stmt.executeQuery(selectQuery);
			if (rs.next()) {
				count = rs.getInt(1);
			}
			rs.close();
			stmt.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR counting entries - " + e.getMessage());
		}
		enqueueConnection(conn);
		return count;
	}

	@Override
	public synchronized void actionCountry(ACTION action, Country country)
			throws DB_Error {
		actionForEntity(action, "country", new String[] { "name" },
				new String[] { "'" + country.escapedGetName() + "'" }, 1, "id",
				country.getId());
	}

	@Override
	public synchronized int countCountries(String countrySubStr, Movie movie)
			throws DB_Error {
		ArrayList<String> tables = new ArrayList<String>();
		ArrayList<String> params = new ArrayList<String>();
		ArrayList<String> args = new ArrayList<String>();
		
		tables.add("country");
		
		if (countrySubStr != null && countrySubStr.length() > 0) {
			params.add("lower(name) LIKE ");
			args.add("'%" + countrySubStr + "%'");
		}
		if (movie != null) {
			tables.add("country_movie");
			params.add("country_id = "); args.add("id");
			params.add("movie_id = "); args.add(Integer.toString(movie.getId()));
		}

		return countGeneric(
				ListToArray(tables), 
				ListToArray(params),
				ListToArray(args), 
				"id");
	}

	@Override
	public synchronized List<Country> getCountries(String countrySubStr,
			int from, int to) throws DB_Error {
		return Selector.getCountries(countrySubStr, from, to, this);
	}

	@Override
	public synchronized void actionActor(ACTION action, Actor actor) throws DB_Error {
		actionForEntity(
				action,
				"actor",
				new String[] { "first_name", "last_name", "Sex" },
				new String[] { "'" + actor.escapedGetFirstName() + "'",
						"'" + actor.escapedGetLastName() + "'",
						Integer.toString(actor.getSex().getValue()) }, 3, "id",
				actor.getId());
	}

	@Override
	public synchronized int countActors(String firstNameSubStr, String lastNameSubStr,
			SEX sex, Movie movie) throws DB_Error {
		String sexValue;
		
		ArrayList<String> tables = new ArrayList<String>();
		ArrayList<String> params = new ArrayList<String>();
		ArrayList<String> args = new ArrayList<String>();
		
		tables.add("actor");
		if (firstNameSubStr != null && firstNameSubStr.length()>0){
			params.add("lower(first_name) LIKE "); args.add("'%" + firstNameSubStr + "%'");
		}
		if (lastNameSubStr != null && lastNameSubStr.length()>0){
			params.add("lower(last_name) LIKE "); args.add("'%" + lastNameSubStr + "%'");
		}
		if (sex != null) {
			sexValue = Integer.toString(sex.getValue());
			params.add("Sex="); args.add(sexValue);
		}
		
		
		// if movie is null we want only the actor table

		if (movie!= null){
			tables.add("actor_movie");
			params.add("actor_id = "); args.add("id");
			params.add("movie_id = "); args.add(Integer.toString(movie.getId()));
		}
		
		return countGeneric(
				ListToArray(tables),
				ListToArray(params),
				ListToArray(args)
				, "id");
	}

	@Override
	public synchronized List<Actor> getActors(String firstNameSubStr, String lastNameSubStr,
			SEX sex, Movie movie, int from, int to) throws DB_Error {
		return Selector.getActors(firstNameSubStr, lastNameSubStr, sex, movie,
				from, to, this);
	}

	@Override
	public synchronized List<Movie> getActorMovies(Actor actor, int from, int to)
			throws DB_Error {
		return Selector.getMoviesForEntity(actor.getId(), from, to, this,
				"actor");
	}

	@Override
	public synchronized void deleteMovieFromActor(Actor actor, Movie movie) throws DB_Error {
		deleteGeneric("actor_movie", new String[] { "actor_id", " movie_id" },
				new int[] { actor.getId(), movie.getId() });
	}

	@Override
	public synchronized void actionDirector(ACTION action, Director director)
			throws DB_Error {
		actionForEntity(action, "director", new String[] { "first_name",
				"last_name" },
				new String[] { "'" + director.escapedGetFirstName() + "'",
						"'" + director.escapedGetLastName() + "'" }, 2, "id",
				director.getId());
	}

	@Override
	public synchronized int countDirectors(String firstNameSubStr, String lastNameSubStr,
			Movie movie) throws DB_Error {
		ArrayList<String> tables = new ArrayList<String>();
		ArrayList<String> params = new ArrayList<String>();
		ArrayList<String> args = new ArrayList<String>();
		
		tables.add("director");
		
		if (firstNameSubStr != null && firstNameSubStr.length()>0){
			params.add("lower(first_name) LIKE "); args.add("'%" + firstNameSubStr + "%'");
		}
		
		if (lastNameSubStr != null && lastNameSubStr.length()>0){
			params.add("lower(last_name) LIKE "); args.add("'%" + lastNameSubStr + "%'");
		}

		
		// if movie is null we want only the actor table

		if (movie!= null){
			tables.add("director_movie");
			params.add("director_id = "); args.add("id");
			params.add("movie_id = "); args.add(Integer.toString(movie.getId()));
		}
		
		return countGeneric(
						ListToArray(tables),
						ListToArray(params),
						ListToArray(args),
						"id");
	}

	@Override
	public synchronized List<Director> getDirectors(String firstNameSubStr,
			String lastNameSubStr, Movie movie, int from, int to)
			throws DB_Error {
		return Selector.getDirectors(firstNameSubStr, lastNameSubStr, movie,
				from, to, this);
	}

	@Override
	public synchronized List<Movie> getDirectorMovies(Director director, int from, int to)
			throws DB_Error {
		return Selector.getMoviesForEntity(director.getId(), from, to, this,
				"director");
	}

	@Override
	public synchronized void deleteMovieFromDirector(Director director, Movie movie)
			throws DB_Error {
		deleteGeneric("director_movie", new String[] { "director_id",
				" movie_id" }, new int[] { director.getId(), movie.getId() });
	}

	@Override
	public synchronized void actionMovie(ACTION action, Movie movie) throws DB_Error {
		actionForEntity(
				action,
				"movie",
				new String[] { "name", "release_year", "rank" },
				new String[] { "'" + movie.escapedGetTitle() + "'",
						Integer.toString(movie.getReleaseYear()),
						Float.toString(movie.getRank()) }, 2, "id",
				movie.getId());
	}

	public synchronized String genreListToString(List<Genre> genres) {
		if (genres == null || genres.size() == 0)
			return "()";
		StringBuilder gen = new StringBuilder();
		gen.append('(');
		for (Genre g : genres) {
			gen.append(g.getId());
			gen.append(',');
		}
		gen.setCharAt(gen.length() - 1, ')');
		return gen.toString();
	}

	private String[] ListToArray(List<String> l) {
		String[] arr = new String[l.size()];
		int i = 0;
		for (String s : l) {
			arr[i] = s;
			i++;
		}
		return arr;
	}

	@Override
	public synchronized int countMovies(String titleSubStr, int releaseYearMin,
			int releaseYearMax, float rankMin, float rankMax, Actor actor,
			Director director, List<Genre> geners, Country country)
			throws DB_Error {
		ArrayList<String> tables = new ArrayList<String>();
		ArrayList<String> params = new ArrayList<String>();
		ArrayList<String> values = new ArrayList<String>();

		tables.add("movie");
		if (titleSubStr != null && titleSubStr.length() > 0) {
			params.add("movie.name LIKE ");
			values.add("'%" + titleSubStr + "%'");
		}
		if (releaseYearMin > -1) {
			params.add("release_year >= ");
			values.add(Integer.toString(releaseYearMin));
		}
		if (releaseYearMax > -1) {
			params.add("release_year <= ");
			values.add(Integer.toString(releaseYearMax));
		}
		if (rankMin > -1) {
			params.add("rank >= ");
			values.add(Float.toString(rankMin));
		}
		if (rankMax > -1) {
			params.add("rank <= ");
			values.add(Float.toString(rankMax));
		}

		if (actor != null) {
			tables.add("actor_movie");
			params.add("actor_movie.movie_id =");
			values.add("movie.id");
			params.add("actor_id = ");
			values.add(Integer.toString(actor.getId()));
		}

		if (director != null) {
			tables.add("director_movie");
			params.add("director_movie.movie_id =");
			values.add("movie.id");
			params.add("director_id = ");
			values.add(Integer.toString(director.getId()));
		}

		if (country != null) {
			tables.add("country_movie");
			params.add("country_movie.movie_id =");
			values.add("movie.id");
			params.add("country_id = ");
			values.add(Integer.toString(country.getId()));
		}

		if (geners != null && geners.size() > 0) {
			String genersL = genreListToString(geners);
			String generLength = Integer.toString(geners.size());
			String subQuery = "(select COUNT(genre_id) from genre_movie where movie.id = movie_id AND genre_id IN "
					+ genersL + ")";
			params.add(generLength + " IN "); values.add(subQuery);
		}
		return countGeneric(ListToArray(tables), ListToArray(params),
				ListToArray(values), "id");
	}

	@Override
	public synchronized List<Movie> getMovies(String titleSubStr, int releaseYearMin,
			int releaseYearMax, float rankMin, float rankMax, Actor actor,
			Director director, List<Genre> geners, Country country, int from,
			int to) throws DB_Error {
		return Selector.getMovies(titleSubStr, releaseYearMin, releaseYearMax,
				rankMin, rankMax, actor, director, geners, country, from, to,
				this);
	}

	@Override
	public synchronized List<Director> getMovieDirectors(Movie movie, int from, int to)
			throws DB_Error {
		return Selector.getMovieDirectors(movie, from, to, this);
	}

	@Override
	public synchronized List<Actor> getMovieActors(Movie movie, int from, int to)
			throws DB_Error {
		return Selector.getMovieActors(movie, from, to, this);
	}

	@Override
	public synchronized List<Genre> getMovieGenres(Movie movie, int from, int to)
			throws DB_Error {
		return Selector.getMovieGenres(movie, from, to, this);
	}

	@Override
	public synchronized List<Country> getMovieCountries(Movie movie, int from, int to)
			throws DB_Error {
		return Selector.getMovieCountries(movie, from, to, this);
	}

	@Override
	public synchronized void actionEntityFromMovie(ACTION action, Movie movie,
			ENTITIES entity, int entityId) throws DB_Error {
		// get table name
		String table;
		switch (entity) {
		case ACTOR:
			table = "actor";
			break;
		case COUNTRY:
			table = "country";
			break;
		case DIRECTOR:
			table = "director";
			break;
		default:
			table = "genre";// entity == ENTITIES.GENRE
		}
		// do action
		switch (action) {
		case CREATE:
			insertGeneric(
					table + "_movie",
					new String[] { "movie_id", table + "_id" },
					new String[] { Integer.toString(movie.getId()),
							Integer.toString(entityId) }, 2);
			break;
		case DELETE:
			deleteGeneric(table + "_movie", new String[] { "movie_id",
					table + "_id" }, new int[] { movie.getId(), entityId });
			break;
		case UPDATE:
			break;// Update has no meaning here
		}
	}

	@Override
	public synchronized List<DBQuestion> getQuestions(PATTERN pattern,
			String titleSubStr, int from, int to) throws DB_Error {
		int startResults = from;
		int lengthOfResults = to - from;
		Connection conn = dequeueConnection();
		Statement stmt1, stmt2;
		ResultSet rs1, rs2;
		List<DBQuestion> questions = new ArrayList<DBQuestion>();

		String tableAnswer = "";
		switch (pattern) {
		case ACTOR_PLAYED_IN_MOVIE: {
			tableAnswer = "actor";
			break;
		}
		case GENRE_OF_MOVIE: {
			tableAnswer = "genre";
			break;
		}
		}

		String r = String.format("ans_%s_r", tableAnswer);
		String w1 = String.format("ans_%s_w1", tableAnswer);
		String w2 = String.format("ans_%s_w2", tableAnswer);
		String w3 = String.format("ans_%s_w3", tableAnswer);

		try {
			stmt1 = conn.createStatement();
			stmt2 = conn.createStatement();
			String where = "";
			if (titleSubStr != null)
				where += " AND movie.name LIKE '%" + titleSubStr + "%'";
			rs1 = stmt1.executeQuery("SELECT * FROM "
					+ pattern.getQueryHelper() + where + " LIMIT "
					+ startResults + "," + lengthOfResults);

			while (rs1.next()) {
				// select the actual names of the actors
				rs2 = stmt2.executeQuery("SELECT * FROM " + tableAnswer
						+ " WHERE id IN (" + rs1.getInt(r) + ", "
						+ rs1.getInt(w1) + ", " + rs1.getInt(w2) + ", "
						+ rs1.getInt(w3) + ")");

				String movieName = rs1.getString("movie.name");
				int id = rs1.getInt(pattern.getTableName() + ".id");

				// setting the names of the actors for the answers
				String[] answers = new String[4];
				int rightAnswerIndex = -1;
				int i = 0;
				while (rs2.next()) {
					if (rs2.getInt("id") == rs1.getInt(r)) {
						rightAnswerIndex = i;
					}
					if (tableAnswer.equals("actor")) {
						answers[i] = rs2.getString("first_name");
						answers[i] += " " + rs2.getString("last_name");
					} else {
						answers[i] = rs2.getString("name");
					}
					i++;
				}
				rs2.close();

				questions.add(new DBQuestion(id, pattern, rs1.getInt("level"),
						movieName, rightAnswerIndex, answers));
			}
			rs1.close();
			stmt2.close();
			stmt1.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR get questions - " + e.getMessage());
		}

		enqueueConnection(conn);
		return questions;
	}

	@Override
	public synchronized List<DBQuestion> getQuestions(PATTERN pattern,
			String titleSubStr1, String titleSubStr2, int from, int to)
			throws DB_Error {
		int startResults = from;
		int lengthOfResults = to - from;
		Connection conn = dequeueConnection();
		Statement stmt1, stmt2;
		ResultSet rs1, rs2;
		List<DBQuestion> questions = new ArrayList<DBQuestion>();

		String r = "ans_movie_r";
		String w1 = "ans_movie_w1";
		String w2 = "ans_movie_w2";
		String w3 = "ans_movie_w3";

		try {
			stmt1 = conn.createStatement();
			stmt2 = conn.createStatement();
			String where = "";
			if (pattern.equals(PATTERN.MOVIE_RELEASED_IN_YEAR)) {
				where += "";
				if (titleSubStr1 != null && titleSubStr1.length()>0)
					where += " WHERE year>=" + titleSubStr1;
				if (titleSubStr2 != null && titleSubStr2.length()>0) {
					if (titleSubStr1 == null)
						where += " WHERE ";
					else
						where += " AND ";
					where += "year<=" + titleSubStr2;
				}
			} else {
				if (titleSubStr1 != null)
					where += " AND first_name LIKE '%" + titleSubStr1 + "%'";
				if (titleSubStr2 != null) {
					where += " AND last_name LIKE '%" + titleSubStr2 + "%'";
				}
			}

			rs1 = stmt1.executeQuery("SELECT * FROM "
					+ pattern.getQueryHelper() + where + " LIMIT "
					+ startResults + "," + lengthOfResults);

			while (rs1.next()) {
				// select the actual names of the movies
				rs2 = stmt2.executeQuery("SELECT * FROM movie WHERE id IN ("
						+ rs1.getInt(r) + ", " + rs1.getInt(w1) + ", "
						+ rs1.getInt(w2) + ", " + rs1.getInt(w3) + ")");

				String Name;
				if (pattern.equals(PATTERN.MOVIE_RELEASED_IN_YEAR)) {
					Name = Integer.toString(rs1.getInt("year"));
				} else {
					Name = rs1.getString("first_name") + " "
							+ rs1.getString("last_name");
				}
				int id = rs1.getInt(pattern.getTableName() + ".id");

				// setting the names of the actors for the answers
				String[] answers = new String[4];
				int rightAnswerIndex = -1;
				int i = 0;
				while (rs2.next()) {
					if (rs2.getInt("id") == rs1.getInt(r)) {
						rightAnswerIndex = i;
					}

					answers[i] = rs2.getString("name");

					i++;
				}
				rs2.close();

				questions.add(new DBQuestion(id, pattern, rs1.getInt("level"),
						Name, rightAnswerIndex, answers));
			}
			rs1.close();
			stmt2.close();
			stmt1.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR get questions - " + e.getMessage());
		}

		enqueueConnection(conn);
		return questions;
	}

	@Override
	public synchronized void createQuestion(PATTERN pattern, int entityIdQ, int entityIdA, int level) throws DB_Error {
		Connection conn = dequeueConnection();
		Statement stmt;
		ResultSet rs;

		try {
			stmt = conn.createStatement();

			String answerTable = "";
			String questionField = "";
			String notin = "";
			String answerField = "";
			String top = "";
			String idInitial = "";
			if (entityIdA == -1)
				top="top_";
			switch (pattern) {
			case ACTOR_PLAYED_IN_MOVIE:
				answerTable = "top_actor_movie";
				answerField = "DISTINCT actor_id";
				idInitial = "actor_";
				questionField = "movie_id";
				notin = "NOT IN (SELECT actor_id FROM " + top + "actor_movie WHERE movie_id=" + entityIdQ + ")";
				break;
			case GENRE_OF_MOVIE:
				answerTable = "genre";
				answerField = "id";
				questionField = "movie_id";
				notin = "NOT IN (SELECT genre_id FROM genre_movie WHERE movie_id=" + entityIdQ + ")";
				break;
			case MOVIE_RELEASED_IN_YEAR:
				answerTable = "top_movie";
				answerField = "id";
				questionField = "year";
				notin = "NOT IN (SELECT id FROM " + top + "movie WHERE release_year=" + entityIdQ + ")";
				break;
			case MOVIE_ACTOR_IN:
				answerTable = "top_movie";
				answerField = "id";
				questionField = "actor_id";
				notin = "NOT IN (SELECT movie_id FROM " + top + "actor_movie WHERE actor_id=" + entityIdQ + ")";
				break;
			case MOVIE_DIRECTED_BY:
				answerTable = "top_movie";
				answerField = "id";
				questionField = "director_id";
				notin = "NOT IN (SELECT movie_id FROM " + top + "director_movie WHERE director_id=" + entityIdQ + ")";
				break;
			}

			// check if question already exist
			if (entityIdA != -1 && !pattern.equals(PATTERN.MOVIE_RELEASED_IN_YEAR)){
				rs = stmt.executeQuery("SELECT " + questionField + " FROM "
						+ pattern.getTableName() + " WHERE " + questionField + "="
						+ entityIdQ);
				if (rs.next()) {
					enqueueConnection(conn);
					rs.close();
					stmt.close();
					throw new DB_Error("Question already exist", false);
				}
				rs.close();
			}
			
			rs = stmt.executeQuery("SELECT " + answerField + " FROM "+answerTable+" WHERE "+idInitial+"id "+notin+" ORDER BY RAND() LIMIT 3");
			int countAnswers = 0;
			int[] answers = new int[3];
			while (rs.next()) {
				answers[countAnswers++] = rs.getInt(1);
			}
			if (countAnswers < 3) {
				rs.close();
				stmt.close();
				enqueueConnection(conn);
				throw new DB_Error("Not enough wrong answers are available in the database in order to create the question", false);
			}
			rs.close();

			if (entityIdA == -1) {
				// creating the right answer
				String inQuery = notin.substring(8, notin.length() - 1).concat(" ORDER BY RAND() LIMIT 1");
				rs = stmt.executeQuery(inQuery);
				if (rs.next()) {
					entityIdA = rs.getInt(1);
				} else {
					rs.close();
					stmt.close();
					enqueueConnection(conn);
					throw new DB_Error("There is no right answer for this question", false);
				}
				rs.close();
			}
			
			if (pattern.equals(PATTERN.MOVIE_RELEASED_IN_YEAR)){
				//check if question made for movie_release_in_year is not unique
				rs = stmt.executeQuery("SELECT " + questionField + " FROM "
						+ pattern.getTableName() + " WHERE year="
						+ entityIdQ + " && ans_movie_r=" + entityIdA);
				if (rs.next()) {
					enqueueConnection(conn);
					rs.close();
					stmt.close();
					throw new DB_Error("Question already exist", false);
				}
				rs.close();
			}

			// insert new question to the database
			stmt.executeUpdate("INSERT INTO " + pattern.getTableName()
					+ " VALUES(null," + level + "," + entityIdQ + ","
					+ entityIdA + "," + answers[0] + "," + answers[1] + ","
					+ answers[2] + ")");

			stmt.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR creating new question - "
					+ e.getMessage());
		}
		enqueueConnection(conn);
	}

	@Override
	public synchronized void deleteQuestion(PATTERN pattern, int questionId)
			throws DB_Error {
		Connection conn = dequeueConnection();
		Statement stmt;

		try {
			stmt = conn.createStatement();

			stmt.executeUpdate("DELETE FROM " + pattern.getTableName()
					+ " WHERE id=" + questionId);

			stmt.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR deleting question - " + e.getMessage());
		}
		enqueueConnection(conn);
	}

	@Override
	public synchronized List<Actor> getAnswersActors(PATTERN pattern,
			int entityIdQ, int from, int to) throws DB_Error {

		int startResults = from;
		int lengthOfResults = to - from;
		Connection conn = dequeueConnection();
		Statement stmt;
		ResultSet rs;
		List<Actor> actors = new ArrayList<Actor>();

		try {
			stmt = conn.createStatement();

			rs = stmt
					.executeQuery("SELECT * FROM actor,actor_movie WHERE actor.id=actor_id AND movie_id="
							+ entityIdQ
							+ " LIMIT "
							+ startResults
							+ ","
							+ lengthOfResults);
			while (rs.next()) {
				Actor actor = new Actor(rs.getInt("actor.id"),
						rs.getString("first_name"), rs.getString("last_name"),
						rs.getInt("Sex") == 1 ? SEX.MALE : SEX.FEMALE);
				actors.add(actor);
			}
			rs.close();
			stmt.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR getting possible actor answers - "
					+ e.getMessage());
		}
		enqueueConnection(conn);
		return actors;
	}

	@Override
	public synchronized List<Movie> getAnswersMovies(PATTERN pattern,
			int entityIdQ, int from, int to) throws DB_Error {
		int startResults = from;
		int lengthOfResults = to - from;
		Connection conn = dequeueConnection();
		Statement stmt;
		ResultSet rs;
		List<Movie> movies = new ArrayList<Movie>();

		try {
			stmt = conn.createStatement();

			String query = "";
			switch (pattern) {
			case MOVIE_RELEASED_IN_YEAR:
				query = "SELECT * FROM movie WHERE release_year=" + entityIdQ;
				break;
			case MOVIE_ACTOR_IN:
				query = "SELECT * FROM movie WHERE id IN (SELECT movie_id FROM actor_movie WHERE actor_id="
						+ entityIdQ + ")";
				break;
			case MOVIE_DIRECTED_BY:
				query = "SELECT * FROM movie,director_movie WHERE movie.id=movie_id AND director_id="
						+ entityIdQ;
				break;
			}
			query += " LIMIT " + startResults + "," + lengthOfResults;
			rs = stmt.executeQuery(query);
			while (rs.next()) {
				Movie movie = new Movie(rs.getInt("movie.id"),
						rs.getString("movie.name"), rs.getInt("release_year"),
						rs.getFloat("rank"), rs.getInt("votes"));
				movies.add(movie);
			}
			rs.close();
			stmt.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR getting possible movie answers - "
					+ e.getMessage());
		}
		enqueueConnection(conn);
		return movies;
	}

	@Override
	public synchronized List<Genre> getAnswersGenres(PATTERN pattern,
			int entityIdQ, int from, int to) throws DB_Error {
		int startResults = from;
		int lengthOfResults = to - from;
		Connection conn = dequeueConnection();
		Statement stmt;
		ResultSet rs;
		List<Genre> genres = new ArrayList<Genre>();

		try {
			stmt = conn.createStatement();

			rs = stmt
					.executeQuery("SELECT * FROM genre,genre_movie WHERE genre.id=genre_id AND movie_id="
							+ entityIdQ
							+ " LIMIT "
							+ startResults
							+ ","
							+ lengthOfResults);
			while (rs.next()) {
				Genre genre = new Genre(rs.getInt("genre.id"),
						rs.getString("genre.name"));
				genres.add(genre);
			}
			rs.close();
			stmt.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("ERROR getting possible genre answers - "
					+ e.getMessage());
		}
		enqueueConnection(conn);
		return genres;
	}

	public synchronized int getMovieMinYear() {
		return movieMinYear;
	}

	public synchronized int getMovieMaxYear() {
		return movieMaxYear;
	}

	@Override
	public synchronized int countAnswers(PATTERN pattern, int entityIdQ) throws DB_Error {
		ArrayList<String> tables = new ArrayList<String>();
		ArrayList<String> params = new ArrayList<String>();
		ArrayList<String> args = new ArrayList<String>();
		String fieldName;
		switch (pattern) {
		case ACTOR_PLAYED_IN_MOVIE:
			tables.add("actor_movie");
			fieldName = "movie_id";
			break;

		case MOVIE_RELEASED_IN_YEAR:
			tables.add("movie");
			fieldName = "release_year";
			break;

		case MOVIE_ACTOR_IN:
			tables.add("actor_movie");
			fieldName = "actor_id";
			break;

		case MOVIE_DIRECTED_BY:
			tables.add("director_movie");
			fieldName = "director_id";
			break;

		default:
			// GENRE_OF_MOVIE:
			tables.add("genre_movie");
			fieldName = "movie_id";
			break;
		}
		params.add(fieldName + " = ");
		args.add(Integer.toString(entityIdQ));

		return countGeneric(ListToArray(tables), ListToArray(params),
				ListToArray(args), fieldName);
	}

	@Override
	public synchronized int countQuestionMovie(PATTERN pattern, String titleSubStr)
			throws DB_Error {

		ArrayList<String> tables = new ArrayList<String>();
		ArrayList<String> params = new ArrayList<String>();
		ArrayList<String> args = new ArrayList<String>();
		tables.add(pattern.getTableName());
		if (titleSubStr != null && titleSubStr.length() > 0) {
			tables.add("movie");
			params.add("movie_id =");
			args.add("movie.id");
			params.add("movie.name Like");
			args.add("'%" + titleSubStr + "%'");
		}
		return countGeneric(ListToArray(tables), ListToArray(params),
				ListToArray(args), "id");
	}

	@Override
	public synchronized int countQuestioPerson(PATTERN pattern, String firstNameSubStr,
			String lastNameSubStr) throws DB_Error {
		ArrayList<String> tables = new ArrayList<String>();
		ArrayList<String> params = new ArrayList<String>();
		ArrayList<String> args = new ArrayList<String>();

		String tableName;

		switch (pattern) {
		case MOVIE_DIRECTED_BY:
			tableName = "director";
			break;
		default:
			tableName = "actor";
		}

		tables.add(pattern.getTableName());
		if (firstNameSubStr != null && firstNameSubStr.length() > 0) {
			tables.add(tableName);
			params.add(tableName + "_id =");
			args.add(tableName + ".id");
			params.add(tableName + ".first_name Like");
			args.add("'%" + firstNameSubStr + "%'");
		}
		if (lastNameSubStr != null && lastNameSubStr.length() > 0) {
			if (!tables.contains(tableName)) {
				tables.add(tableName);
				params.add(tableName + "_id =");
				args.add(tableName + ".id");
			}
			params.add(tableName + ".last_name Like");
			args.add("'%" + lastNameSubStr + "%'");
		}

		return countGeneric(ListToArray(tables), ListToArray(params),
				ListToArray(args), "id");
	}

	@Override
	public synchronized int countQuestioYear(PATTERN pattern, int minYear, int maxYear)
			throws DB_Error {
		ArrayList<String> tables = new ArrayList<String>();
		ArrayList<String> params = new ArrayList<String>();
		ArrayList<String> args = new ArrayList<String>();

		tables.add(pattern.getTableName());
		if (minYear != -1) {
			params.add("year >= ");
			args.add(Integer.toString(minYear));
		}
		if (maxYear != -1) {
			params.add("year <= ");
			args.add(Integer.toString(maxYear));
		}

		return countGeneric(ListToArray(tables), ListToArray(params),
				ListToArray(args), "id");
	}

	public synchronized void generateQuestions(int questionNum) throws DB_Error {
		for (int level = 0; level < DB.numOfLevels; level++) {
			for (PATTERN p : PATTERN.values()) {
				ArrayList<Integer> questionsIds = new ArrayList<Integer>();
				for (int i = 0; i < questionNum; i++) {
					questionsIds.add(createUniqeQuestions(p, level + 1));
				}

				for (Integer qId : questionsIds) {
					try {
						createQuestion(p, qId, -1, level + 1);
					} catch (DB_Error e) {
						// question was not created
					}
				}
			}
		}
	}

	private Integer createUniqeQuestions(PATTERN pattern, int level)
			throws DB_Error {
		String query = "";
		String levelString;
		int questionsId = -1;

		switch (level) {
		case 1:
			levelString = " release_year >= " + EasyLevelYear;
			break;
		case 2:
			levelString = " release_year > " + MediumLevelYear
					+ "  AND  release_year < " + EasyLevelYear;
			break;
		default:
			levelString = " release_year <= " + MediumLevelYear;
			break;
		}


		Connection conn = dequeueConnection();
		ResultSet rs;
		try {
			Statement stmt = conn.createStatement();

			switch (pattern) {
			case ACTOR_PLAYED_IN_MOVIE:
				query = "SELECT DISTINCT movie_id FROM top_actor_movie WHERE "
						+ levelString 
						+ " AND movie_id NOT IN (SELECT movie_id FROM " + pattern.getTableName()+ ")";

				break;
			case MOVIE_RELEASED_IN_YEAR:
				query = "SELECT release_year FROM top_movie WHERE " + levelString;
				break;
			case MOVIE_ACTOR_IN:
				query = "SELECT actor_id " + "FROM top_actor_movie WHERE "
						+ levelString
						+ " AND actor_id NOT IN (SELECT actor_id FROM " + pattern.getTableName() + ")";

				break;
			case MOVIE_DIRECTED_BY:
				query = "SELECT director_id "
						+ "FROM top_director_movie WHERE "
						+ levelString
						+ " AND director_id NOT IN (SELECT director_id FROM " + pattern.getTableName() + ")";

				break;
			case GENRE_OF_MOVIE:
				query = "SELECT id FROM top_movie WHERE "
						+ levelString
						+ " AND id NOT IN (SELECT movie_id FROM " + pattern.getTableName() + ")";
			}
			query += " ORDER BY RAND() LIMIT " + 1;

			rs = stmt.executeQuery(query);
			while (rs.next()) {
				questionsId = rs.getInt(1);
			}

			rs.close();
			stmt.close();
		} catch (SQLException e) {
			enqueueConnection(conn);
			throw new DB_Error("Error creating random questions - "
					+ e.getMessage());
		}
		enqueueConnection(conn);
		return questionsId;
	}

	@Override
	public synchronized void setDBFilesPaths(String moviesFilePath, String ratingsFilePath,
			String countriesFilePath, String genresFilePath,
			String actorsFilePath, String actressesFilePath,
			String directorsFilePath) {
		parser.setMoviesFilePath(moviesFilePath);
		parser.setRatingsFilePath(ratingsFilePath);
		parser.setCountriesFilePath(countriesFilePath);
		parser.setGenresFilePath(genresFilePath);
		parser.setActorsFilePath(actorsFilePath);
		parser.setActressesFilePath(actressesFilePath);
		parser.setDirectorsFilePath(directorsFilePath);
	}

	@Override
	public synchronized void importToDB() throws DB_Error {
		parser.clearAllParams();
		parser.changeTablesToMyISAM();
		parser.deleteAll();
		parser.parseAllFiles();
		parser.changeTablesToInnoDB();
		parser.findTopMoviesAndActors();
	}

	@Override
	public synchronized void smallUpdateDB() throws DB_Error {
		parser.clearAllParams();
		parser.parseAllFiles();
		parser.findTopMoviesAndActors();
	}

	@Override
	public synchronized void largeUpdateDB() throws DB_Error {
		parser.clearAllParams();
		parser.changeTablesToMyISAM();
		parser.parseAllFiles();
		parser.changeTablesToInnoDB();
		parser.findTopMoviesAndActors();
	}

	@Override
	public synchronized void clearParserParams() {
		parser.clearAllParams();
	}
}